package com.study.spark.scala.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 可更新状态的
  * @author stephen
  * @create 2019-03-20 09:56
  * @since 1.0.0
  */
object StatefulWordCount {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("StatefulWordCount")
    val ssc = new StreamingContext(conf, Seconds(10))

    // 如果使用stateful的算子，必须要设置checkpoint
    // 生产环境中，checkpoint设置在HDFS某个文件夹中
    ssc.checkpoint("./ckp")

    // 创建SocketInputDStream
    val lines = ssc.socketTextStream("localhost", 9999)

    // 进行统计
    val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
    // 更新数据，执行action
    result.updateStateByKey(updateFunction _).print()
    // 可简写为
    //result.updateStateByKey{case (seq,opt)=> {Option(opt.getOrElse(0)+seq.sum)}}.print()

    ssc.start()
    ssc.awaitTermination()

  }

  /**
    * 根据当前的数据和以前的数据去做更新
    *
    * @param newValues    当前的数据，相同key的value都在该序列
    * @param runningCount 以前的数据
    * @return
    */
  def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val current = newValues.sum
    val pre = runningCount.getOrElse(0)
    Option(current + pre)
  }
}
